package msgcenter

import (
	"context"
	"sync"
	"time"

	"gitee.com/doraemon1996/bunny/container"
	"gitee.com/doraemon1996/bunny/goroutine"
	"gitee.com/doraemon1996/bunny/log"
	"gitee.com/doraemon1996/bunny/protocol"
	"gitee.com/doraemon1996/bunny/structure"
)

// 消息中心
// 实现消息分发、
type msgCenter struct {
	structure.Control

	// 阻塞调用结构缓存池
	waitPool *structure.CachePool[*callTimeout]

	// Call等待响应表
	mu         sync.RWMutex
	timeoutMap map[uint64]*callTimeout
}

// 创建消息中心实例对象
func New() MessageCenter {
	return new(msgCenter)
}

// 初始化
func (mc *msgCenter) Init(ctx context.Context) {
	mc.Control.Init(ctx, func() {
		mc.waitPool = structure.NewCachePool[*callTimeout](func() any {
			return new(callTimeout)
		})
		mc.timeoutMap = make(map[uint64]*callTimeout)
	})
}

// 启动
func (mc *msgCenter) Start() {
	mc.Control.Start(func() {
		mc.Add(1)
		goroutine.Submit(func() {
			defer mc.Done()
			mc.timeout()
		})
	})
}

// 停止
func (mc *msgCenter) Stop() {
	mc.Control.Stop(func() {
		mc.Wait()

		mc.mu.Lock()
		timeoutMap := mc.timeoutMap
		mc.timeoutMap = nil
		mc.mu.Unlock()

		for _, elem := range timeoutMap {
			elem.Cancel(ErrMessageCenterClosed)
		}
	})
}

// 检测消息等待超时
func (mc *msgCenter) timeout() {
	ticker := time.NewTicker(time.Second)
	defer ticker.Stop()

	var now int64

	for {
		select {
		case <-mc.CtxDone():
			return
		case <-ticker.C:
			now = time.Now().Unix()
			mc.mu.RLock()
			for _, cw := range mc.timeoutMap {
				if cw.CheckTimeout(now) {
					cw.Cancel(ErrMessageTimeout)
				}
			}
			mc.mu.RUnlock()
		}
	}
}

// 发送消息到指定服务
func (mc *msgCenter) Send(source, destination uint64, msgID int32, msgBody []byte) {
	if mc.Closed() {
		log.Errorf("Send %d from %X to %X, but message center closed.", msgID, source, destination)
		return
	}

	// 区分

	msg := new(protocol.Message)
	msg.Type = protocol.MessageType_Send
	msg.Source = source
	msg.Destination = destination
	msg.Sequence = 0
	msg.MsgId = msgID
	msg.MsgBody = msgBody

	msgCtx := newContext(mc.Context(), msg, nil)
	container.PutMessage(destination, msgCtx)
}

// 发送消息到指定服务，并等待响应
// ti 是超时时间，当 ti > 0 时才生效超时机制，否则将一直等待响应
func (mc *msgCenter) Call(msg container.Message) (data []byte, err error) {
	if mc.Closed() {
		err = ErrMessageCenterClosed
		return
	}

	container.PutMessage(0, msg)

	return
}

// 响应消息
func (mc *msgCenter) Response(sequence uint64, msgBody []byte) (err error) {
	if mc.Closed() {
		return ErrMessageCenterClosed
	}

	mc.mu.Lock()
	wait, ok := mc.timeoutMap[sequence]
	mc.mu.Unlock()

	if ok {
		err = wait.Response(msgBody)
	} else {
		err = ErrMessageTimeout
	}

	return
}
